package com.ojoin.trade.supplyloan_core.web.support;

import java.io.Serializable;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.jms.Destination;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Queues;
import com.ojoin.trade.common.utils.LogUtils;
import com.ojoin.trade.common.utils.NamedThreadFactory;
import com.ojoin.trade.common.utils.SpringUtils;
import com.ojoin.trade.common.web.domain.Pair;
import com.ojoin.trade.supplyloan_core.web.service.QueueSenderServie;

/**
 * 发送消息到ActiveMQ
 */
public class ActiveMQSender implements Runnable {
  private static final Logger LOG = LoggerFactory.getLogger(ActiveMQSender.class);
  private static final ActiveMQSender INSTANCE = new ActiveMQSender();
  private final BlockingQueue<Pair<Destination,Serializable>> queue = Queues.newArrayBlockingQueue(1000);
  private ExecutorService executor;
   
  private QueueSenderServie sender=SpringUtils.getBean(QueueSenderServie.class);
  
  private boolean running;

  private ActiveMQSender() {
    NamedThreadFactory factory = new NamedThreadFactory("ActiveMQ-sender", true);
    executor = Executors.newSingleThreadExecutor(factory);
    executor.submit(this);
     
    Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
      @Override
      public void run() {
        running = false;
        executor.shutdown();
      }
    }));
  }

  public static ActiveMQSender getInstance() {
    return INSTANCE;
  }


  boolean doSend(Destination destination,  Serializable message) {
	  return sender.send(destination, message);
  }

  @Override
  public void run() {
    running = true;
    while (running) {
    	Pair<Destination,Serializable> msg;
      try {
        msg = queue.take();
      } catch (InterruptedException e) {
        LogUtils.error(LOG,"ActiveMQSender interrupted, still "+queue.size()+" in queue ,now exit ");
        return;
      }
      if (msg == null || sender == null) {
        continue;
      }
      boolean sendOk=doSend(msg.first,msg.second);
      if(!sendOk){
    	  queue.add(msg);
      }
    }
  }

  public void asyncSend(Destination destination, Serializable m) {
	  Pair<Destination,Serializable> msg=Pair.of(destination, m);
      queue.add(msg);
  }
}
